from pyspark.sql import SQLContext, SparkSession
from pyspark.sql import functions as F
from passwd import dbuser, dbpass, dbhost, dbport
import random
import re
if __name__ == '__main__':
    # spark 初始化
    spark = SparkSession. \
        Builder(). \
        appName('sql'). \
        master('local'). \
        getOrCreate()
    # mysql 配置(需要修改)
    prop = {'user': dbuser, 
            'password': dbpass,
            'driver': 'com.mysql.jdbc.Driver'}
    # database 地址(需要修改)
    url = 'jdbc:mysql://' + dbhost + ':' + dbport + '/ratemap'
    # 读取表，data1是用户表，data2是business表
    data1 = spark.read.jdbc(url=url, table='user', properties=prop)
    data2 = spark.read.jdbc(url=url, table='business', properties=prop)
    data3 = spark.read.jdbc(url=url, table='checkin', properties=prop)

    ### 合并business表和checkin表，根据business_id字段
    data = data2.join(data3, data2.business_id == data3.business_id, 'inner')
    ### 获取保存合并后的表的所有数据的数组
    info = data.collect()
    ### 字典d1，用来保存商店名称，对应的上座率
    d1 = {}
    ### 字典d2，用来保存商店名称，对应的城市
    d2 = {}
    ### 对info数组进行循环遍历
    for i in info:
        arr = i.date.split(', ')
        ### 保存店铺对应的上座率
        d1[i.name] = len(arr)
        ### 保存店铺对应的城市
        d2[i] = i.city
    ### 以下代码，将所有城市，都放入一个数组中，并且这些城市不重复
    list = data2.collect()
    cities = []
    for item in list:
        if (item.city not in cities):
            cities.append(item.city)
    ### 创建一个store变量，保存城市对应的上座率最高的商店
    store = ''
    ### 创建一个max变量，保存最高的上座率
    max = 0
    ### 字典d3，保存城市对应的上座率最高的商店
    d3 = {}
    ### 循环遍历所有城市
    for city in cities:
        ### 再循环遍历d2
        for key, value in d2.items():
            ### 如果商店属于该城市city
            if value == city:
                ### 并且该商店上座率更高
                if d1[key.name] > max:
                    ### 更新上座率和商店
                    max = d1[key.name]
                    store = key
        ### 保存该城市对应的上座率最高的商店
        d3[city] = store
        ### 更新变量
        max = 0
        store = ''
    ### 用户输入的城市
    area = 'Henderson'
    print(d3[area])
    
    # 关闭spark会话
    spark.stop()